package com.hb.test

import io.reactivex.Observable
import io.reactivex.rxkotlin.subscribeBy
import io.reactivex.schedulers.Schedulers
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit

class Order3 {
    val threadCount=Runtime.getRuntime().availableProcessors()
    //线程池
    val threadPoolExecutor= Executors.newFixedThreadPool(threadCount)
    //调度器
    val scheduler = Schedulers.from(threadPoolExecutor)

    fun getAddressFromDB3(userId:Long):Observable<Address>{
        //返回的是一个Observable<Goods>类型的数据
        return Observable.defer{
            Thread.sleep(1000)
            Observable.just(Address(userId,"杭州"))
        }
    }

    fun getGoodsFromDB3(goodsId:Long):Observable<Goods>{
        //返回的是一个Observable<Goods>类型的数据
        return Observable.defer{
            Thread.sleep(1000)
            Observable.just(Goods(goodsId,"好吃的食物",20))
        }
    }


    fun doOrder3(goods: Goods,address: Address):Long{
        //下单操作
        Thread.sleep(1000)
        return 1L
    }

    /**
     * rxkotlin
     * 1. 将异步编程变得优雅、直观、不用对每个异步请求执行一个回调，同时还可以组合多个异步任务
     * 2. 不需要书写多线程代码，值需要指定响应的策略便可以使用多线程的功能
     * 3. 响应式编程的特点就是数据流的处理，简单来说就是将数据处理的过程变得像流水线一样，比如
     * A==>B==>C后继者不需要阻塞等待结果，而是由前一个矗立着将结果通知他。
     *
     * 场景：
     *  假设下单之后，需要给商家及消费者推送消息，那么用流处理如下表示：
     *      doOrder()
     *          .map(doNotifyCustomer)
     *          .map(doNotifyShop)
     *          .map(doOther)
     *  我们可以独一原始数据进行处理，生成一个新的数据然后传递给下一个矗立着，这些处理过程都是异步非
     *  阻塞的。可以看出，流氏调用相对于会带哦的方式优雅的很多不在需要编写大量的嵌套回调函数，从而使
     *  代码更加简洁易懂。
     */
    fun rxOrder(goodsId: Long,userId: Long){
        var goods:Goods?=null
        var address:Address?=null

        //方法在指定调度器中执行
        val goodsFromDB = getGoodsFromDB3(1).subscribeOn(scheduler)
        val addressFromDB = getAddressFromDB3(1).subscribeOn(scheduler)

        //合并两个Observable
        Observable.merge(goodsFromDB,addressFromDB).subscribeBy(
           onNext = {
               when(it) {
                   is Goods -> {
                       println("Goods....")
                       goods=it
                   }
                   is Address -> {
                       println("Address....")
                       address=it
                   }
               }
           }
        ,
        onComplete = {
            doOrder3(goods!!,address!!)
        });
    }

}
fun main() {

    val start = System.currentTimeMillis()

    val order3 = Order3()
    order3.rxOrder(1L,2L);
   // order(1L,2L)

    //Thread.sleep(2000)
    val end = System.currentTimeMillis()
    val toSeconds = TimeUnit.MILLISECONDS.toSeconds(end - start)
    println("耗时： $toSeconds s")
}
